package com.chb.weblog.hive.mr.pre;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SmallFileCombiner
{
	public static void main(String[] args)
			throws Exception
	{
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(SmallFileCombiner.class);

		job.setMapperClass(SmallFileCombinerMapper.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		job.setInputFormatClass(CombineTextInputFormat.class);
		CombineTextInputFormat.setMaxInputSplitSize(job, 268435456L);
		//CombineTextInputFormat.setInputPaths(job, new Path[] { new Path(args[0]) });
		//FileOutputFormat.setOutputPath(job, new Path(args[1]));
		CombineTextInputFormat.setInputPaths(job, new Path("file:///F:/Project/HadoopProject/HadoopUtils/src/com/chb/weblog/data/access.log.fensi"));
		FileOutputFormat.setOutputPath(job, new Path("file:///F:/Project/HadoopProject/HadoopUtils/src/com/chb/weblog/data/output"));

		job.setNumReduceTasks(0);

		job.waitForCompletion(true);
	}

	static class SmallFileCombinerMapper extends Mapper<LongWritable, Text, Text, NullWritable>
	{
		NullWritable v = NullWritable.get();

		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
			context.write(value, this.v);
		}
	}
}
